package top.jolyoulu.core.example

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * @Author: JolyouLu
 * @Date: 2024/2/6 16:32
 * @Description
 * 编号	字段名称	字段类型	字段含义
 * 1	date	String	用户点击行为的日期
 * 2	user_id	Long	用户的 ID
 * 3	session_id	String	Session 的 ID
 * 4	page_id	Long	某个页面的 ID
 * 5	action_time	String	动作的时间点
 * 6	search_keyword	String	用户搜索的关键词
 * 7	click_category_id	Long	某一个商品品类的 ID
 * 8	click_product_id	Long	某一个商品的 ID
 * 9	order_category_ids	String	一次订单中所有品类的 ID 集合
 * 10	order_product_ids	String	一次订单中所有商品的 ID 集合
 * 11	pay_category_ids	String	一次支付中所有品类的 ID 集合
 * 12	pay_product_ids	String	一次支付中所有商品的 ID 集合
 * 13	city_id	Long	城市 id
 *
 * 存在问题
 * Q1.使用累加器替代reduceByKey去掉shuffle操作
 */
object Spark01_Example_HotCategoryTop10Analysis_Optimize3 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
    val sc: SparkContext = new SparkContext(sparkConf)

    //读取日志数据
    val path: String = this.getClass.getClassLoader.getResource("datas/user_visit_action.txt").toURI.getPath
    val actionRDD: RDD[String] = sc.textFile(path)

    val acc = new HotCategoryAccumulator()
    sc.register(acc, "HotCategory")

    //循环累计
    actionRDD.foreach(
      action => {
        val datas: Array[String] = action.split("_")
        //点击
        if (datas(6) != "-1") {
          acc.add(datas(6), "click")
          //下单
        } else if (datas(8) != "null") {
          val ids: Array[String] = datas(8).split(",")
          ids.foreach(id => acc.add(id, "order"))
          //支付
        } else if (datas(10) != "null") {
          val ids: Array[String] = datas(10).split(",")
          ids.foreach(id => acc.add(id, "pay"))
        } else {
          Nil
        }
      }
    )
    //获取累加器值
    val accVal: mutable.Map[String, HotCategory] = acc.value
    //取出HotCategory
    val categories: mutable.Iterable[HotCategory] = accVal.map(_._2)
    //排序
    val sort: List[HotCategory] = categories.toList.sortWith(
      (left, right) => {
        if (left.clickCnt > right.clickCnt) {
          true
        } else if (left.clickCnt == right.clickCnt) {
          if (left.orderCnt > right.orderCnt) {
            true
          } else if (left.orderCnt == right.orderCnt) {
            left.payCnt > right.payCnt
          } else {
            false
          }
        } else {
          false
        }
      }
    )
   sort.take(10).foreach(println)

    //关闭环境
    sc.stop()
  }

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  /**
   * 自定义累加器
   */
  class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {

    private val hcMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = {
      hcMap.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
      new HotCategoryAccumulator()
    }

    override def reset(): Unit = {
      hcMap.clear()
    }

    override def add(v: (String, String)): Unit = {
      val cid = v._1
      val actionType = v._2
      val category: HotCategory = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      if (actionType == "click") {
        category.clickCnt += 1
      } else if (actionType == "order") {
        category.orderCnt += 1
      } else if (actionType == "pay") {
        category.payCnt += 1
      }
      hcMap.update(cid, category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      val map1 = this.hcMap
      val map2 = other.value
      map2.foreach {
        case (cid, hc) => {
          val category: HotCategory = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hc.clickCnt
          category.orderCnt += hc.orderCnt
          category.payCnt += hc.payCnt
          map1.update(cid, category)
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = {
      hcMap
    }
  }
}
